package cn.azzhu.proj

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * 分渠道统计用户行为
 *
 * @author azzhu
 * @create 2020-09-25 22:09:01
 */
object AppMarketingByChannel {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    env
      .addSource(new SimulatedEventSource)
      .assignAscendingTimestamps(_.ts)
      .filter(_.behaviour != "UnInstall")
      .map(r => {
        ((r.channel,r.behaviour),1L)
      })
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))
      .process(new MarketingCountByChannel)
      .print()

    env.execute("AppMarketingByChannel")
  }

  class MarketingCountByChannel extends ProcessWindowFunction[((String,String),Long),(String,Long,Long),(String,String),TimeWindow] {
    override def process(key: (String, String), context: Context,
                         elements: Iterable[((String, String), Long)],
                         out: Collector[(String, Long, Long)]): Unit = {
      out.collect((key._1,elements.size,context.window.getEnd))
    }
  }
}
